package com.dd.keel.core.framework.service;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.NoSuchElementException;

import org.apache.commons.pool2.ObjectPool;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.msgpack.MessagePack;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.dd.keel.core.common.utils.StopWatch;
import com.dd.keel.core.framework.model.BodyModel;
import com.dd.keel.core.framework.model.RequestModel;
import com.dd.keel.core.framework.model.ResponseModel;
import com.dd.keel.core.framework.proxy.ProxyConn;
import com.dd.keel.core.framework.proxy.SoaProxyMsgMarshall;
import com.dd.keel.core.framework.proxy.SoaProxyRequest;
import com.dd.keel.core.framework.proxy.SoaProxyResponse;
import com.dd.keel.core.framework.server.NodeConfig;
import com.dd.keel.core.framework.server.ServerEngine;

/**
 * Service invoke util
 * 
 * @author Kevin.XU
 *
 */
public class ServiceInvoke {

	final static private Logger LOGGER = LoggerFactory
			.getLogger(ServiceInvoke.class);

	private static GenericObjectPool<ProxyConn> pool;
	private static ServiceInstanceFactory instanceFactory;
	private static NodeConfig nodeConfig;

	public static ObjectPool<ProxyConn> getPool() {
		return pool;
	}

	public void setPool(GenericObjectPool<ProxyConn> pool) {
		ServiceInvoke.pool = pool;
		ServiceInvoke.pool.setNumTestsPerEvictionRun(10);
		ServiceInvoke.pool.setTimeBetweenEvictionRunsMillis(60 * 1000);
		ServiceInvoke.pool.setBlockWhenExhausted(false);
	}

	public static ServiceInstanceFactory getInstanceFactory() {
		return instanceFactory;
	}

	public void setInstanceFactory(ServiceInstanceFactory instanceFactory) {
		ServiceInvoke.instanceFactory = instanceFactory;
	}

	public static NodeConfig getNodeConfig() {
		return nodeConfig;
	}

	public void setNodeConfig(NodeConfig nodeConfig) {
		ServiceInvoke.nodeConfig = nodeConfig;
	}

	/**
	 * local service coordination
	 * 
	 * @param group
	 * @param service
	 * @param version
	 * @return
	 */
	private static String getServiceCoordinate(String group, String service,
			String version) {
		StringBuilder sb = new StringBuilder();
		sb.append(getNodeConfig().getHost()).append('#')
				.append(getNodeConfig().getPort()).append('#')
				.append(group).append('#').append(service).append('#')
				.append(version);
		return sb.toString();
	}

	/**
	 * 
	 * @param group
	 * @param service
	 * @param version
	 * @return
	 */
	private static String getRemoteServiceCoordinate(String group,
			String service, String version) {
		StringBuilder sb = new StringBuilder();
		sb.append("Remote").append('#').append(group).append('#')
				.append(service).append('#').append(version);
		return sb.toString();
	}

	/**
	 * release connection
	 * 
	 * @param conn
	 */
	private static void releaseConnection(ProxyConn conn) {
		try {
			getPool().returnObject(conn);
		} catch (Exception ex) {
			LOGGER.error(ex.getMessage(), ex);
		}
	}

	/**
	 * invoke remote service
	 * 
	 * @param groupName
	 * @param serviceName
	 * @param serviceVersion
	 * @param traceId
	 * @param traceSpanId
	 * @param requestParams
	 * @return
	 */
	public static Map<String,Object> invokeRemoteSoaService(String groupName,
			String serviceName, String serviceVersion, String traceId,
			String traceSpanId, Map<String, Object> requestParams) {
		// construct request model
		Map<String,Object> request = new HashMap<String,Object>();
		request.put("group", groupName);
		request.put("service", serviceName);
		request.put("version", serviceVersion);
		request.put("trace_id", traceId);
		request.put("trace_span_id", traceSpanId);
		if (requestParams != null) {
			request.put("data", requestParams);
		}
		StopWatch stopWatch = new StopWatch();
		stopWatch.start();
		// encode request model
		byte[] raw = null;
		try {
			raw = JSON.toJSONString(request).getBytes("UTF-8");
		} catch (Exception ex) {
			LOGGER.error(ex.getMessage(), ex);
		}
		if (raw == null) {
			return null;
		}
		SoaProxyRequest soaProxyRequest = new SoaProxyRequest();
		soaProxyRequest.setGroupName(groupName);
		soaProxyRequest.setServiceName(serviceName);
		soaProxyRequest.setServiceVersion(serviceVersion);
		soaProxyRequest.setRequestBytes(raw);
		SoaProxyResponse soaProxyResponse = null;
		// require connection
		ProxyConn conn = null;
		try {
			conn = getPool().borrowObject();
		} catch (NoSuchElementException ex) {
			LOGGER.error(ex.getMessage(), ex);
		} catch (IllegalStateException ex) {
			LOGGER.error(ex.getMessage(), ex);
		} catch (Exception ex) {
			LOGGER.error(ex.getMessage(), ex);
		}
		if (conn == null) {
			return null;
		}
		// send request
		boolean success = false;
		int retry = 2;
		while (retry-- > 0) {
			boolean ret = false;
			try {
				ret = SoaProxyMsgMarshall.sendRequest(conn, soaProxyRequest);
				if (!ret) {
					break;
				}
			} catch (Exception ex) {
				LOGGER.error(ex.getMessage(), ex);
				if (retry > 0)
					conn.reconnect();
				continue;
			}
			try {
				soaProxyResponse = SoaProxyMsgMarshall.receiveResponse(conn);
			} catch (Exception ex) {
				LOGGER.error(ex.getMessage(), ex);
				if (retry > 0)
					conn.reconnect();
				continue;
			}
			success = true;
			break;
		}
		if (!success) {
			releaseConnection(conn);
			return null;
		}
		if (soaProxyResponse.getStatus() == 1
				|| soaProxyResponse.getResponseBytes() == null) {
			releaseConnection(conn);
			return null;
		}
		// decode response model
		Map<String,Object> response = null;
		try {
			response = (Map<String,Object>)JSON.parse(new String(soaProxyResponse.getResponseBytes(),"UTF-8"));
		} catch (Exception ex) {
			LOGGER.error(ex.getMessage(), ex);
		}
		releaseConnection(conn);
		stopWatch.stop();
		StringBuilder dapperLog = new StringBuilder();
		dapperLog
				.append("DAPPER::: ")
				.append(traceId)
				.append(",")
				.append(traceSpanId)
				.append(",")
				.append(getRemoteServiceCoordinate(groupName, serviceName,
						serviceVersion)).append(" ; cost time (ms) = ")
				.append(stopWatch.cost());
		LOGGER.info(dapperLog.toString());
		return response;
	}

	/**
	 * invoke local service
	 * 
	 * @param groupName
	 * @param serviceName
	 * @param serviceVersion
	 * @param traceId
	 * @param traceSpanId
	 * @param requestParams
	 * @return
	 */
	public static Map<String,Object> invokeLocalSoaService(String groupName,
			String serviceName, String serviceVersion, String traceId,
			String traceSpanId, Map<String, Object> requestParams) {
		ISoaService instance = getInstanceFactory().getInstance(groupName,
				serviceName, serviceVersion);
		if (instance == null) {
			return null;
		}
		String currentTraceSpanId = getServiceCoordinate(groupName,
				serviceName, serviceVersion);
		StopWatch stopWatch = new StopWatch();
		stopWatch.start();
		Map<String,Object> response = instance.execute(traceId, currentTraceSpanId,
				requestParams);
		stopWatch.stop();
		StringBuilder dapperLog = new StringBuilder();
		dapperLog.append("DAPPER::: ").append(traceId).append(",")
				.append(traceSpanId).append(",").append(currentTraceSpanId)
				.append(" ; cost time (ms) = ").append(stopWatch.cost());
		LOGGER.info(dapperLog.toString());
		return response;
	}

	/**
	 * invoke service
	 * 
	 * @param groupName
	 * @param serviceName
	 * @param serviceVersion
	 * @param traceId
	 * @param traceSpanId
	 * @param requestParams
	 * @return
	 */
	public static Map<String,Object> invokeSoaService(String groupName,
			String serviceName, String serviceVersion, String traceId,
			String traceSpanId, Map<String, Object> requestParams) {
		ISoaService instance = getInstanceFactory().getInstance(groupName,
				serviceName, serviceVersion);
		Map<String,Object> response = null;
		if (instance == null) {
			response = invokeRemoteSoaService(groupName, serviceName,
					serviceVersion, traceId, traceSpanId, requestParams);
		} else {
			response = invokeLocalSoaService(groupName, serviceName,
					serviceVersion, traceId, traceSpanId, requestParams);
		}
		return response;
	}

}
